package io.task.message.socket.task;

import io.task.message.socket.*;
import io.task.message.socket.thread.pool.QueueModeSynchronousTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Setter
@Getter
public class SendTask extends QueueModeSynchronousTask<Packet> {
    private ChannelContext channelContext = null;
    private Config config = null;
    private Handler handler = null;


    private ConcurrentLinkedQueue<Packet> forSendAfterSslHandshakeCompleted = null;                                            //new ConcurrentLinkedQueue<>();
    public boolean canSend = true;

    public ConcurrentLinkedQueue<Packet> getForSendAfterSslHandshakeCompleted(boolean forceCreate) {
        if (forSendAfterSslHandshakeCompleted == null && forceCreate) {
            synchronized (this) {
                if (forSendAfterSslHandshakeCompleted == null) {
                    forSendAfterSslHandshakeCompleted = new ConcurrentLinkedQueue<>();
                }
            }
        }

        return forSendAfterSslHandshakeCompleted;
    }


    public SendTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        this.config = channelContext.config;
        this.handler = config.getTioHandler();


        getMsgQueue();
    }

    @Override
    public boolean addMsg(Packet packet) {
        if (this.isCanceled()) {
            log.info("{}, 任务已经取消，{}添加到发送队列失败", channelContext);
            return false;
        }


        {
            return msgQueue.add(packet);
        }
    }

    /**
     * 清空消息队列
     */
    @Override
    public void clearMsgQueue() {
        Packet p = null;
        forSendAfterSslHandshakeCompleted = null;
        while ((p = msgQueue.poll()) != null) {
            try {
                channelContext.processAfterSent(p, false);
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }
        }
    }

    private ByteBuffer getByteBuffer(Packet packet) {
        try {
            ByteBuffer byteBuffer = packet.getPreEncodedByteBuffer();
            if (byteBuffer == null) {
                byteBuffer = handler.encode(packet, config, channelContext);
            }

            if (!byteBuffer.hasRemaining()) {
                byteBuffer.flip();
            }
            return byteBuffer;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }


    private static final int MAX_CAPACITY_MIN = Packet.MAX_DATA_LENGTH - 1024;    //减掉1024是尽量防止溢出的一小部分还分成一个tcp包发出
    private static final int MAX_CAPACITY_MAX = MAX_CAPACITY_MIN * 10;


    @Override
    public void runTask() {
        if (msgQueue.isEmpty()) {
            return;
        }

        int queueSize = msgQueue.size();
        if (queueSize == 1) {
            //			System.out.println(1);
            Packet packet = msgQueue.poll();
            if (packet != null) {
                sendPacket(packet);
            }
            return;
        }

        int listInitialCapacity = Math.min(queueSize, canSend ? 300 : 1000);

        Packet packet = null;
        List<Packet> packets = new ArrayList<>(listInitialCapacity);
        List<ByteBuffer> byteBuffers = new ArrayList<>(listInitialCapacity);

        int allBytebufferCapacity = 0;

        boolean sslChanged = false;
        while ((packet = msgQueue.poll()) != null) {
            ByteBuffer byteBuffer = getByteBuffer(packet);

            packets.add(packet);
            byteBuffers.add(byteBuffer);

            allBytebufferCapacity += byteBuffer.limit();


            if ((canSend && allBytebufferCapacity >= MAX_CAPACITY_MIN) || (allBytebufferCapacity >= MAX_CAPACITY_MAX) || sslChanged) {
                break;
            }
        }


        if (allBytebufferCapacity == 0) {
            return;
        }
        ByteBuffer allByteBuffer = ByteBuffer.allocate(allBytebufferCapacity);
        for (ByteBuffer byteBuffer : byteBuffers) {
            allByteBuffer.put(byteBuffer);
        }

        allByteBuffer.flip();


        this.sendByteBuffer(allByteBuffer, packets);

    }

    public boolean sendPacket(Packet packet) {
        ByteBuffer byteBuffer = getByteBuffer(packet);


        sendByteBuffer(byteBuffer, packet);
        return true;
    }

    /**
     * @param byteBuffer
     * @param packets    Packet or List<Packet>
     * @author tanyaowu
     */
    public void sendByteBuffer(ByteBuffer byteBuffer, Object packets) {
        if (byteBuffer == null) {
            log.error("{},byteBuffer is null", channelContext);
            return;
        }

        if (!IoUtils.checkBeforeIO(channelContext)) {
            return;
        }

        //		if (!byteBuffer.hasRemaining()) {
        //			byteBuffer.flip();
        //		}

        ReentrantLock lock = channelContext.writeCompletionHandler.lock;
        lock.lock();
        try {
            canSend = false;
            WriteCompletionHandler.WriteCompletionHandlerVo writeCompletionHandlerVo = new WriteCompletionHandler.WriteCompletionHandlerVo(byteBuffer, packets);
            channelContext.asynchronousSocketChannel.write(byteBuffer, writeCompletionHandlerVo, channelContext.writeCompletionHandler);
            channelContext.writeCompletionHandler.condition.await();
        } catch (InterruptedException e) {
            log.error(e.toString(), e);
        } finally {
            lock.unlock();
        }
    }


    private final Queue<Packet> msgQueue = new LinkedBlockingQueue<>();

    @Override
    public Queue<Packet> getMsgQueue() {

        return msgQueue;
    }

}
